package com.func.wordcount;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Hello world!
 *
 */
public class App extends Configured implements Tool {

	public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			IntWritable one = new IntWritable(1);
			while (tokenizer.hasMoreTokens()) {
				Text word = new Text(tokenizer.nextToken());
				context.write(word, one);
			}
		}

	}

	public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
				int sum = 0;
				for(IntWritable num:values) {
					sum += num.get();
				}
				context.write(key, new IntWritable(sum));
		}

	}

	@Override
	public int run(String[] strings) throws Exception {
		String inputPath = "hdfs://localhost:9000/input";
		String outputPath = "hdfs://localhost:9000/output";
		Job job = Job.getInstance(getConf(),"wordcount");
		job.setJarByClass(App.class);
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextInputFormat.addInputPath(job,new Path(inputPath));
		TextOutputFormat.setOutputPath(job,new Path(outputPath));
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		boolean success = job.waitForCompletion(true);
		return success?0:1;
	}

	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new App(),args);
		System.exit(ret);
	}
}
